Module design patterns

This page collects design patterns for modules that interact with a third-party app or a DE tool: how to report what happened, how to make repeated runs safe, and how to check the target is ready before writing to it.

Both bring conventions your function has to bridge to the platform's file-based contract. A third-party app has its own data schema, its own object identifiers, and its own rules for what makes a valid record. A DE tool has its own API and its own error handling, which rarely map cleanly onto a job that either succeeded or failed.

The examples are Python, matching stari module scaffold --type python-module, but nothing here depends on the language.

Always return outputs, even when the function fails

The user can only read what the job uploaded. A function that raises out of its entry point leaves them with a failed job and nothing to open — the most expensive failure mode to support, because every occurrence becomes a request for someone with agent host access to go read a local log.

Declare logs and errors as required outputs in the manifest:

"outputs": [
  { "name": "requirements", "type": "file", "required": true, "upload_as": "model" },
  { "name": "logs", "type": "file", "required": true, "upload_as": "artifact" },
  { "name": "errors", "type": "file", "required": true, "upload_as": "artifact" }
]

Then catch everything at the entry point and return the output list either way:

def push_requirements(input_json: str, temp_dir: str) -> list[Output]:
    """Never raises -- failures are captured in logs.txt / errors.txt
    so the job always completes with visible output."""
    temp = Path(temp_dir)
    log_handler, err_handler = add_file_handlers(temp)   # logs.txt at INFO, errors.txt at ERROR
    output_path = temp / "requirements.json"

    try:
        function_input = PushRequirementsInput.model_validate_json(input_json)
        process(function_input, output_path)
    except Exception:
        logger.exception("PushRequirements failed")
        if not output_path.exists():
            output_path.write_text(json.dumps(empty_document()), encoding="utf-8")

    flush_and_detach(log_handler, err_handler)
    return [
        Output(name="requirements", type=OutputType.FILE, path=str(output_path)),
        Output(name="logs", type=OutputType.FILE, path=str(temp / "logs.txt")),
        Output(name="errors", type=OutputType.FILE, path=str(temp / "errors.txt")),
    ]

Two file handlers on the root logger — logs.txt at INFO, errors.txt at ERROR — give the reader a full trace and a short "what went wrong" without having to skim. An empty errors.txt is a meaningful result: the job ran clean.

Never overwrite a good revision with a failed run

When a function's output is uploaded as a new revision of the input model, the failure path needs care. Writing a placeholder {} so the job can complete would commit an empty revision over content someone depends on — the job reports success and the model is gone.

Fall back to the input revision instead, so a failed run is a no-op rather than a deletion:

def preserve_input_on_failure(input_model_path: Path, output_path: Path) -> None:
    """Copy the input revision to output so a failed run does not wipe the model."""
    if input_model_path.is_file() and input_model_path.stat().st_size > 0:
        output_path.write_text(input_model_path.read_text(encoding="utf-8"), encoding="utf-8")
    else:
        output_path.write_text("{}", encoding="utf-8")

The new revision is byte-identical to the previous one, errors.txt explains why, and the reader can compare revisions to confirm nothing changed.

tip

The agent names an uploaded model revision from the basename of the output file you return. Default the output filename to the input revision's filename, so repeated runs keep one consistent name instead of renaming the model every time.

Make repeated runs converge

Jobs get repeated — by someone who did not see the first one finish, by a scheduled sync, by someone testing. If a repeat duplicates records in the third-party app, every one of those is an incident, and it is discovered by the app's users rather than by you.

Derive object identity from stable keys rather than letting the app assign one, then update-or-create:

def record_id(project_id: str, qualified_name: str) -> str:
    """Same input path always yields the same UUID, so re-runs converge."""
    return str(uuid.uuid5(NAMESPACE, f"{project_id}::{qualified_name}"))

With uuid5 over a stable path plus an upsert write, running the same sync five times leaves the app in the same state as running it once. That property is what lets you tell someone "just run it again" when a job fails halfway — without it, every partial failure needs manual cleanup before the retry.

The same idea applies to documents you write back to the platform: stamp a $schema key into every document your module produces. The next run can then recognize its own output instead of guessing from the shape, which is what makes a round trip re-runnable.

Default writes to a dry run

A dry_run parameter that defaults to true lets someone point a new function at real data and see exactly what it would do, with no possibility of damage.

The pattern only works if the dry run is informative, which means the payload is written either way:

document = build_document(project_name, payloads)
output_path.write_text(json.dumps(document, indent=2), encoding="utf-8")

if post:
    return sync_to_app(client, document["requirements"])

logger.info("DRY RUN -- payloads generated but NOT posted to the app")
return {"created": 0, "updated": 0, "unchanged": 0}

Reviewing the payload is the entire point, so write it before the branch, not inside the if. The reader inspects the artifact on the job, then re-runs with dry_run=false.

Apply the same default-to-safe rule to anything your module can create in the third-party app. Where creating structure is unambiguous — a folder that only has to exist, a tag created empty — put it behind an explicit input that defaults to off:

{
  "create_missing_folders": {
    "type": "parameter",
    "validation_types": ["@boolean"],
    "optional": true
  }
}

Where it is not unambiguous, do not offer it at all. Structure in a third-party app usually encodes decisions your module has no basis to make — what the category levels mean, which taxonomy the organization agreed on. A module that invents them produces records that look right to the module and wrong to everyone using the app, and the mistake surfaces long after the job succeeded.

Check preconditions before the first write

A precondition is anything that must already be true in the third-party app for your writes to land correctly. Check them all before the first write, and order the checks so the cheap ones fail first. A job that dies on a local JSON parse costs nothing; a job that dies halfway through a hierarchy upsert leaves records someone has to clean up by hand.

Order	Check	Why it comes here
1	Input payload shape	Local, instant, no credentials needed
2	Credentials present	Fails without a network round trip
3	Target container resolves (project, program, workspace)	One cheap API call; catches the most common mistake
4	Required structure exists (categories, tiers, levels)	Determines whether records can be created at all
5	Cross-references resolve	Catches functions that were run in the wrong order

Report every missing credential at once

Credentials for a third-party app belong in the agent's environment, not in job inputs, so a misconfigured agent is a common failure. Report every missing variable in one message, and say where they belong — reporting them one at a time turns a single fix into four job runs.

missing = [name for name, value in required_vars if not value]
if missing:
    raise RuntimeError(
        f"Missing credentials: {', '.join(missing)}. "
        "Set them as environment variables on the agent host, "
        "or in a .env file next to the module binary."
    )

Log the presence of each variable as set or MISSING — never the value. That single log line resolves most credential tickets without anyone having to reproduce the job.

warning

Credentials read from the agent environment apply to every job that agent runs. When a function should act as the user who submitted the job rather than as the agent, declare an authentication input instead and let the platform deliver the sign-in. See Authenticating from a module.

Check the required structure, not just existence

Resolving a project by name proves the project exists. It does not prove it can hold the records you are about to create. Third-party apps commonly have a structural prerequisite — category levels, a required taxonomy, a root node — that a new project does not have yet, and whose absence surfaces as an opaque API error on your first write.

Check it explicitly, and name the setup step in the error:

def validate_tier_levels(tiers: dict[int, str], project_name: str) -> None:
    """Three levels are required: Program -> Assembly -> Part."""
    if len(tiers) >= 3:
        return
    raise ValueError(
        f"Project {project_name!r} must have at least three equipment tiers "
        f"(used as Program -> Assembly -> Part). "
        f"The API returned {len(tiers)} tier(s) from equipment-tiers/byProject. "
        f"Create three equipment tier levels for this project, then retry."
    )

Write errors the reader can act on

The person reading your error message is looking at a failed job in the web app. They do not have your source, your log level, or your mental model of the third-party app. Assume the message is all they get.

Every failure message should answer three questions:

1. What was checked — name the object and the constraint, not the function that raised.
2. What was observed — the actual count, name, or value that failed the check.
3. What to do next — the concrete action, and where it has to happen.

The third part is the one most modules omit, and the one that decides whether the message saves a support round trip:

Instead of	Write
KeyError: 'equipmentTierId'	Project 'Ares' must have at least three equipment tiers (used as Program -> Assembly -> Part). The API returned 1 tier(s). Create three equipment tier levels for this project, then retry.
Project not found	Project 'Ares ' not found. Available: Ares, Ares Test, Helios.
ValidationError: 12 errors	12 equipment reference(s) missing: A-100, A-101, ... (and 7 more). Sync equipment first with @acme:push_equipment.

Two habits produce most of that improvement:

• List the valid alternatives when a lookup fails. Name lookups fail constantly, and almost always for the same reasons — a trailing space, a near-duplicate project, an environment holding different data than the one the reader was looking at. You already hold the list you searched, so print it.
• Truncate long lists but keep the count. 12 missing: A-100, A-101, ... (and 7 more) stays readable when a whole hierarchy is absent, and still conveys the scale.

Finally, keep "nothing to do" separate from "something is wrong". A query that legitimately returns zero rows is not a failure, and raising on it teaches readers to distrust your errors. Succeed, and spend a WARNING on the diagnosis they are about to need — most often, that they picked a name that exists but is not the one holding the data.

Checklist

Before publishing a module that writes to a third-party app:

• The entry point never raises; logs and errors are required outputs.
• A failed run preserves the previous revision instead of committing a placeholder.
• Object IDs come from stable keys, so a repeat converges instead of duplicating.
• dry_run defaults to true, and the payload is written in both modes.
• Anything the module creates is behind an input that defaults to off.
• Preconditions are checked in cost order, and all of them run before the first write.
• Every missing credential is reported in one message.
• Every error says what was checked, what was observed, and what to do next.
• Failed lookups list the valid alternatives.

Related pages

• Module concepts — the file-based contract, lifecycle, and packaging conventions these patterns build on.
• Authenticating from a module — acting as the job's user instead of the agent.
• Module Manifest API Reference — input types, optional, required, and upload_as.
• Integration 302 — Build a Python module — a worked build of the scaffold these examples assume.